package com.flute.haflute.monitor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.jobbox.common.CloudContext;
import com.flute.haflute.jobbox.master.CloudJob;
import com.flute.haflute.jobbox.master.JobTracker;
import com.flute.haflute.net.RMINetClient;

/**
 * JOB监控告警线程
 * 
 */
public class JobMonitor extends Thread {

	private static Logger logger = LoggerFactory.getLogger(JobMonitor.class);

	private boolean active = true;
	private boolean isAgentMonitor = false;
	private int jobmonitorsleeptime = 5000;
	Object clusterJobMonitor;
	List<CloudJob> cachedJobList;
	int errorCout = 0;

	public void init() {

	}

	public JobMonitor() {
		this.setName("JobMonitor");
		this.isAgentMonitor = false;
	}

	public JobMonitor(boolean isAgent) {
		this.isAgentMonitor = isAgent;
	}

	public void run() {
		while (active) {
			try {
				init();
				if (!isAgentMonitor) {
					getRunningJobs();
				} else {
					// 发送系统参数
					getSystemAbility();
				}

				Thread.sleep(jobmonitorsleeptime);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public void getRunningJobs() throws Exception {
		List<CloudJob> runningCloudJobs = JobTracker.getInstance()
				.runningJobList();

		// 需要比较 JOB 只要不同就发送
		boolean changed = compareToLast(runningCloudJobs);
		if (changed) {
			setParamToSend("setJobInfo", makeJobInfo());
		}
	}

	/**
	 * 发送JOB消息时 设置发送参数
	 * 
	 * @param joblist
	 */
	private void setParamToSend(String invokename, List<JobMonitorInfo> joblist) {
		if (joblist == null) {
			return;
		}
		Map<String, List<JobMonitorInfo>> map = new HashMap<String, List<JobMonitorInfo>>();
		map.put(ClusterContext.getLocalAddress().getHostAddress(), joblist);
		MessageCyst msgCst = new MessageCyst();
		msgCst.setParammap(map);
		msgCst.setInvokemethod(invokename);
		sendMsg(msgCst);
	}

	private void setParamToSend(String invokename, SystemAbility systemability) {
		Map<String, SystemAbility> map = new HashMap<String, SystemAbility>();
		map.put(ClusterContext.getLocalAddress().getHostAddress(),
				systemability);
		MessageCyst msgCst = new MessageCyst();
		msgCst.setParammap(map);
		msgCst.setInvokemethod(invokename);
		sendMsg(msgCst);
	}

	private void sendMsg(MessageCyst msgCst) {
		if (msgCst.getInvokemethod() == null || msgCst.getParammap() == null) {
			return;
		}
		String refId = ClusterConstants.CLUSTER_MONITOR_REF;
		try {
			if (clusterJobMonitor == null) {
				clusterJobMonitor = RMINetClient.getServiceObject(refId);
				logger.info("RMINetClient getServiceObject success ");
				errorCout = 0;
			}
			RMINetClient.invoke(clusterJobMonitor, msgCst.getInvokemethod(),
					msgCst.getParammap());
		} catch (Exception e) {
			if (errorCout == 0) {
				logger.error("sendMsg error ", e);
				clusterJobMonitor = null;
			}
			errorCout++;
		}

	}

	/**
	 * 比较对象（CollectionUtils）
	 * 
	 * @param cloudJoblist
	 *            new
	 * @return change true, no change false
	 * @throws Exception
	 */
	private boolean compareToLast(List<CloudJob> cloudJoblist) throws Exception {
		boolean changed = false;
		if (cachedJobList == null || cloudJoblist == null) {
			if (cachedJobList != null || cloudJoblist != null)
				changed = true;
		} else if (cachedJobList.size() != cloudJoblist.size()) {
			changed = true;
		} else {
			cloudJoblist.removeAll(cachedJobList);
			if (cloudJoblist.size() > 0) {
				changed = true;
			}
		}

		if (changed)
			cachedJobList = cloudJoblist;

		return changed;
	}

	private List<JobMonitorInfo> makeJobInfo() throws Exception {

		List<JobMonitorInfo> joblist = new ArrayList<JobMonitorInfo>();

		if (cachedJobList == null) {
			return joblist;
		}

		Iterator<CloudJob> it = cachedJobList.iterator();
		while (it.hasNext()) {
			CloudJob job = it.next();
			List<String> childs = CloudContext.getChids();

			JobMonitorInfo jminfo = new JobMonitorInfo(ClusterContext
					.getLocalAddress().getHostAddress());
			jminfo.setJobid(job.getJobRequestMessage().getJobConf().getJobId());
			jminfo.setJobDescription(job.getJobRequestMessage().getJobConf()
					.getJobDescription());
			jminfo.setJobstatus(job.getStatus());
			jminfo.setChilds(childs);
			jminfo.setReceiveDate(job.createTime);
			joblist.add(jminfo);
		}

		return joblist;

	}

	public void getSystemAbility() {
		SystemAbility systemability = new SystemAbility();
		Memory memory = Memory.getInstance();
		Cpu cpu = Cpu.getInstance();
		systemability.setFreePhysicalMemorySize(memory
				.getFreePhysicalMemorySize() / (1024 * 1024));
		systemability.setJVMMemoryRate(memory.getJVMMemoryRate());
		systemability.setTotalPhysicalMemorySize(memory
				.getTotalPhysicalMemorySize() / (1024 * 1024));
		systemability.setTotalCpuCores(cpu.getTotalCpuCores(-1));
		setParamToSend("setSystemInfo", systemability);
	}
}
